use aspect_codec::prelude::*;
use aspect_db_mq_types::*;

#[derive(Clone, Debug)]
pub struct QueueState {
    /// The index of the next message that will be written to the queue.
    next_idx: u64,

    /// The index of the oldest message in the queue table.  If this is the same
    /// as `next_idx`, then the message doesn't actually exist and the queue is
    /// empty.
    tail_idx: u64,

    /// All the reader IDs, in sorted order.
    readers: Vec<ReaderId>,

    /// Reader positions, the message indexes they last read.  Indexes
    /// correspond to reader ID list.
    reader_positions: Vec<u64>,
}

impl QueueState {
    /// Creates a new instance using a given list of initial readers.  Sorts the
    /// provided list.
    pub fn new_with_readers(mut readers: Vec<ReaderId>) -> Self {
        readers.sort();
        let reader_positions = vec![0; readers.len()];
        Self {
            next_idx: 0,
            tail_idx: 0,
            readers,
            reader_positions,
        }
    }

    /// Returns the "next index", or the idx assigned to the next message to be
    /// written to the queue.
    pub fn next_idx(&self) -> u64 {
        self.next_idx
    }

    /// Returns the "tail index", or the index of the oldest message still in
    /// the queue.  If this is the same as the next index value, then that
    /// message actually doesn't exist and the queue is empty.
    pub fn tail_idx(&self) -> u64 {
        self.tail_idx
    }

    /// Returns if the queue is empty.
    pub fn is_empty(&self) -> bool {
        self.next_idx == self.tail_idx
    }

    /// Sanity checks the queue state to ensure that it's well-formed.  If this
    /// ever fails for a queue that's already in the database then shit is
    /// fucked.
    pub fn sanity_check(&self) -> Result<(), SanityError> {
        if !self.readers.windows(2).all(|w| w[0] <= w[1]) {
            return Err(SanityError::ReadersUnsorted);
        }

        if self.readers.len() != self.reader_positions.len() {
            return Err(SanityError::ReaderVecsMismatch(
                self.readers.len(),
                self.reader_positions.len(),
            ));
        }

        if self.tail_idx() > self.next_idx() {
            return Err(SanityError::NextTailOob(self.next_idx(), self.tail_idx()));
        }

        for i in 0..self.readers.len() {
            let p = self.reader_positions[i];
            if p < self.tail_idx() || p > self.next_idx() {
                return Err(SanityError::PositionOob(
                    p,
                    self.readers[i],
                    self.tail_idx(),
                    self.next_idx(),
                ));
            }
        }

        Ok(())
    }

    /// Mutates this copy of the state, incrementing the next msg idx, returning
    /// the old value that should be used for a message.
    ///
    /// Obviously, does not flush the state, this must be done yourself.
    pub fn advance_idx(&mut self) -> u64 {
        let idx = self.next_idx;
        self.next_idx += 1;
        idx
    }

    /// Gets the position of a reader.
    pub fn get_reader_position(&self, id: &ReaderId) -> Result<u64, Error> {
        let Ok(i) = self.readers.binary_search(&id) else {
            return Err(Error::ReaderNotPresent(*id));
        };

        Ok(self.reader_positions[i])
    }

    /// Updates the reader position to a new value that's greater or equal to
    /// the current valid.  Idempotent, so safe to call repeatedly with the same
    /// value.  Returns the old value and the new value.
    ///
    /// Does not refresh the tail, so that must be called separately.
    pub fn update_reader_position(
        &mut self,
        id: &ReaderId,
        target: u64,
    ) -> Result<(u64, u64), Error> {
        let Ok(i) = self.readers.binary_search(id) else {
            return Err(Error::ReaderNotPresent(*id));
        };

        let old = self.reader_positions[i];

        let real_target = u64::min(target, self.next_idx());

        if real_target > old {
            self.reader_positions[i] = real_target;
        }

        Ok((old, real_target))
    }

    /// Refreshes the stored tail idx.  Returns the old value and the
    /// new/current value.  If the values are the same, then no change has been
    /// made.
    pub fn refresh_tail(&mut self) -> (u64, u64) {
        let tbl_min = self
            .reader_positions
            .iter()
            .min()
            .copied()
            .unwrap_or(self.next_idx());

        let old = self.tail_idx;
        if tbl_min > old {
            self.tail_idx = tbl_min;
        }

        (old, tbl_min)
    }

    /// Tries to insert a reader, returning `true` if successful or `false` if
    /// they already existed.  They will start reading at the head, unable to
    /// read previously sent messages.
    pub fn try_insert_reader(&mut self, reader_id: &ReaderId) -> bool {
        let insert_idx = match self.readers.binary_search(reader_id) {
            Ok(_) => return false,
            Err(i) => i,
        };

        let read_idx = self.next_idx();
        self.readers.insert(insert_idx, *reader_id);
        self.reader_positions.insert(insert_idx, read_idx);

        true
    }

    /// Tries to remove a reader, returning `true` if successful or `false` if
    /// they were not a reader.  They will not be able to consume any pending
    /// messages they have not already consumed.
    pub fn try_remove_reader(&mut self, reader_id: &ReaderId) -> bool {
        let Ok(reader_idx) = self.readers.binary_search(reader_id) else {
            return false;
        };

        self.readers.remove(reader_idx);
        self.reader_positions.remove(reader_idx);

        true
    }
}

aspect_codec::inst_semimanual_codec! {
    QueueState {
        next_idx field u64,
        tail_idx field u64,
        readers vec ReaderId,
        reader_positions vec u64
    }
}

#[cfg(test)]
mod tests {
    use super::super::id::ReaderId;

    use super::QueueState;

    #[test]
    fn test_qstate_refresh() {
        let r1 = ReaderId::from([0; 16]);
        let r2 = ReaderId::from([1; 16]);
        let r3 = ReaderId::from([2; 16]);
        let mut qs = QueueState::new_with_readers(vec![r1, r2, r3]);

        qs.advance_idx();
        qs.advance_idx();
        qs.advance_idx();

        qs.sanity_check().expect("test: sanity check");

        eprintln!("{qs:?}");

        qs.update_reader_position(&r1, 1)
            .expect("test: update position");
        qs.sanity_check().expect("test: sanity check");
        qs.update_reader_position(&r2, 1)
            .expect("test: update position");
        qs.sanity_check().expect("test: sanity check");
        qs.update_reader_position(&r3, 1)
            .expect("test: update position");
        qs.sanity_check().expect("test: sanity check");

        eprintln!("{qs:?}");

        let (old_tail, new_tail) = qs.refresh_tail();
        qs.sanity_check().expect("test: sanity check");
        assert_eq!(old_tail, 0);
        assert_eq!(new_tail, 1);

        eprintln!("old tail {old_tail}, new tail {new_tail}");
        eprintln!("{qs:?}");

        qs.update_reader_position(&r1, 3)
            .expect("test: update position");
        qs.sanity_check().expect("test: sanity check");
        qs.update_reader_position(&r2, 5)
            .expect("test: update position");
        qs.sanity_check().expect("test: sanity check");
        qs.update_reader_position(&r3, 3)
            .expect("test: update position");
        qs.sanity_check().expect("test: sanity check");

        assert!(qs.reader_positions.iter().all(|i| *i == 3));

        eprintln!("{qs:?}");
    }
}
